package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.FaultToleranceConstant;
import cn.gwm.flink.streaming.constant.VehicleType;
import cn.gwm.flink.streaming.dwd.check.VehicleConsumer;
import cn.gwm.flink.streaming.dwd.consumer.P03PhevConsumer;
import cn.gwm.flink.streaming.dwd.streamsource.CommonStreamSource;
import cn.gwm.flink.streaming.dwd.streamsource.FilterStreamSource;
import cn.gwm.flink.streaming.dwd.streamsource.StreamSource;
import cn.gwm.flink.streaming.dwd.model.DelayEnum;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.flink.streaming.sink.kafka.KafkaConsumerEnum;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

/**
 * @Author: liangguang
 * @Date: 2022/10/11
 * @Description:
 */
public class DwdTestTask {
    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        String jobName = ConfigLoader.get(FaultToleranceConstant.FLINK_JOB_NAME);

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        BaseTask.envSet(env, jobName);
        env.setParallelism(1);


        VehicleConsumer es13TestConsumer = new VehicleConsumer() {
        }.setTopic("ods_test")
                .setGroupId("test")
                .setMileage("VehTotDistance")
                .setCustomType(VehicleType.POWER_TYPE_EV)
                .setVehicleType(VehicleType.ES13)
                .setSource(new FilterStreamSource(new CommonStreamSource(KafkaConsumerEnum.GROUPOFFSETS)) {
                    @Override
                    public DataStream<JSONObject> filter(DataStream<JSONObject> dataStream) {
                        return dataStream.filter(item -> VehicleType.ES13.equals(item.getStr(VehicleType.VEHICLE_TYPE)));
                    }
                });

        VehicleConsumer v71TestConsumer = new VehicleConsumer() {
        }.setTopic("ods_test")
                .setGroupId("test")
                .setMileage("VehTotDistance")
                .setCustomType(VehicleType.POWER_TYPE_PHEV)
                .setVehicleType(VehicleType.V71_PHEV)
                .setSource(new FilterStreamSource(new CommonStreamSource(KafkaConsumerEnum.GROUPOFFSETS)) {
                    @Override
                    public DataStream<JSONObject> filter(DataStream<JSONObject> dataStream) {
                        return dataStream.filter(item -> VehicleType.V71_PHEV.equals(item.getStr(VehicleType.VEHICLE_TYPE)));
                    }
                });

        VehicleConsumer p03TestConsumer = new P03PhevConsumer()
                .setTopic("ods_test")
                .setGroupId("test")
                .setSource(new FilterStreamSource(new CommonStreamSource(KafkaConsumerEnum.GROUPOFFSETS)) {
                    @Override
                    public DataStream<JSONObject> filter(DataStream<JSONObject> dataStream) {
                        return dataStream.filter(item -> VehicleType.P03_PHEV.equals(item.getStr(VehicleType.VEHICLE_TYPE)));
                    }
                });


        DataStream<JSONObject> streamEs13 = es13TestConsumer.handle(env);
        DataStream<JSONObject> streamV71 = v71TestConsumer.handle(env);
        DataStream<JSONObject> streamP03 = p03TestConsumer.handle(env);

        DataStream<JSONObject> union = streamEs13.union(streamV71, streamP03);

//        union.print();

        union.filter(item -> DelayEnum.NORMAL.getCode().equals(item.getStr(BaseFields.DELAY)))
                .addSink(FlinkKafkaUtil.toKafka("dwm_charge_test")).name("toKafkaDwdTest");

        env.execute("dwd-testTask");
    }


}
